package cn.edu.hit.core.combine;

import cn.edu.hit.core.Pair;
import cn.edu.hit.core.message.MessageAdapter;
import com.fasterxml.jackson.databind.ObjectMapper;
import cn.edu.hit.core.Document;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public abstract class CombineAdapterHelper implements CombineAdapter {
    protected static final Logger logger = Logger.getLogger("CombineAdapter");
    protected static final ObjectMapper mapper = new ObjectMapper();

    protected ConcurrentMap<String, Map> configureMap = new ConcurrentHashMap<>();

    @Override
    public void register(String id, Map config) {
        configureMap.put(id, config);
    }

    @Override
    public void clear() {
        configureMap.clear();
    }

    @Override
    public final Stream<Map> service(Pair<Long, Map> pair) {
        long timestamp = pair.key;
        /**
         * arg 是特征Map 其Value 可能是  单值、数组、等
         */
        Map<String, Object> arg = pair.value;
        return configureMap.entrySet().parallelStream().map(entry -> combine(timestamp, arg, entry.getKey(), (Document) entry.getValue()));
    }

    protected final Map combine(long timestamp, Map<String, Object> arg, String id, Document configure) {
        String method = configure.getString(TRANSFER_METHOD);
//            Object transferArg = configure.get(TRANSFER_ARG);

        String combineChannels = configure.getString(COMBINE_CHANENELS);
        List<String> channels = Collections.EMPTY_LIST;
        try {
            channels = mapper.readValue(combineChannels, mapper.getTypeFactory().constructCollectionType(List.class, String.class));
        } catch (IOException e) {
            e.printStackTrace();
        }
        List<Document> channelDocs = channels.stream().map(key -> (Document) arg.getOrDefault(key, null)).filter(o -> o != null).collect(Collectors.toList());

        Document result = new Document();
        result.put(TIMESTAMP, timestamp);
        result.put(FEATURE_ID, id);
        result.put(FEATURE_TYPE, configure.get(FEATURE_TYPE));
        result.put(COMPONENT_NAME, configure.get(COMPONENT_NAME));
        result.put(COMBINE_CHANENELS, channelDocs.parallelStream().flatMap(doc -> {
            if (doc.containsKey(FEATURE_TYPE)) return ((List<String>) doc.get(COMBINE_CHANENELS)).stream();
            else return Stream.of(doc.getString(MessageAdapter.PORT_ID));
        }).collect(Collectors.toList()));

        Object r = combineAndCalculate(channelDocs, configure, method, arg);
        result.put(VALUE, r);
        return result;
    }

    protected abstract Object combineAndCalculate(List<Document> channelDocs, Document configure, String method, Map<String, Object> arg);
}
